package tech.edwardvan.rabbitmqdemo.pattern.routing;

import cn.hutool.core.thread.ThreadUtil;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import tech.edwardvan.rabbitmqdemo.pattern.CommonUtil;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

/**
 * 路由模式-消费者
 *
 * @author EdwardVan
 */
@Slf4j
public class Consumer {

    public static void main(String[] args) {

        Thread consumer1 = new Thread(Consumer::consume1);
        consumer1.setName(CommonUtil.CONSUMER_1);

        Thread consumer2 = new Thread(Consumer::consume2);
        consumer2.setName(CommonUtil.CONSUMER_2);

        consumer1.start();
        consumer2.start();
    }

    public static void consume1() {
        try {
            //获取一个连接
            Connection connection = CommonUtil.getConnection();

            //从连接中获取一个通道
            Channel channel = connection.createChannel();

            channel.basicQos(1);

            //声明交换机
            channel.exchangeDeclare(CommonUtil.EXCHANGE_ROUTING, BuiltinExchangeType.DIRECT, false, true, null);

            /*
            创建队列声明
                1.队列名
                2.是否是持久队列
                3.是否是排他队列(独占)
                4.是否自动删除
                5.其他参数属性设置
            */
            channel.queueDeclare(CommonUtil.QUEUE_ROUTING_1, false, true, true, null);

            //绑定队列到交换机
            channel.queueBind(CommonUtil.QUEUE_ROUTING_1, CommonUtil.EXCHANGE_ROUTING, "error");

            //获取当前线程名称
            String consumerName = Thread.currentThread().getName();

            //定义消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String msg = new String(body, StandardCharsets.UTF_8);
                    log.info("{}收到消息:{}", consumerName, msg);
                    ThreadUtil.sleep(1, TimeUnit.SECONDS);
                    //手动应答
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
            //关闭自动应答
            channel.basicConsume(CommonUtil.QUEUE_ROUTING_1, false, consumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void consume2() {
        try {
            //获取一个连接
            Connection connection = CommonUtil.getConnection();

            //从连接中获取一个通道
            Channel channel = connection.createChannel();

            channel.basicQos(1);

            //声明交换机
            channel.exchangeDeclare(CommonUtil.EXCHANGE_ROUTING, BuiltinExchangeType.DIRECT, false, true, null);

            /*
            创建队列声明
                1.队列名
                2.是否是持久队列
                3.是否是排他队列(独占)
                4.是否自动删除
                5.其他参数属性设置
            */
            channel.queueDeclare(CommonUtil.QUEUE_ROUTING_2, false, true, true, null);
            //绑定队列到交换机
            channel.queueBind(CommonUtil.QUEUE_ROUTING_2, CommonUtil.EXCHANGE_ROUTING, "error");
            channel.queueBind(CommonUtil.QUEUE_ROUTING_2, CommonUtil.EXCHANGE_ROUTING, "info");
            channel.queueBind(CommonUtil.QUEUE_ROUTING_2, CommonUtil.EXCHANGE_ROUTING, "warn");

            //获取当前线程名称
            String consumerName = Thread.currentThread().getName();

            //定义消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String msg = new String(body, StandardCharsets.UTF_8);
                    log.info("{}收到消息:{}", consumerName, msg);
                    ThreadUtil.sleep(1, TimeUnit.SECONDS);
                    //手动应答
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
            //关闭自动应答
            channel.basicConsume(CommonUtil.QUEUE_ROUTING_2, false, consumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
